package ruc.TDEngine.read;

import cn.hutool.core.codec.Base64;
import okhttp3.MediaType;
import okhttp3.Request;
import okhttp3.RequestBody;
import okhttp3.Response;
import ruc.Consts.constant;
import ruc.utils.IOUtils;
import ruc.TDEngine.TDEngine_Common;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.util.Random;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.atomic.AtomicLong;

public class TDEngine_ConcurrentRead {
    private int bound;
    private String IP, port, dbName, tbName,userName,password;

    public TDEngine_ConcurrentRead(String IP, String port, String dbName, String tbName, String userName, String password,int bound) {
        this.IP = IP;
        this.port = port;
        this.dbName = dbName;
        this.tbName = tbName;
        this.bound = bound;
        this.userName = userName;
        this.password = password;
    }

    private Request getRequest(int limit, int offset) {
        String Authorization = Base64.encode(String.format("%s:%s", userName, password).getBytes(StandardCharsets.UTF_8));
        String URL = String.format("http://%s:%s/rest/sql", IP, port);
        String query = String.format("select * from %s.%s limit %d offset %d", dbName, tbName, limit, offset);

        Request request = new Request.Builder()
                .url(URL)
                .method("POST", RequestBody.create(MediaType.parse("text/plain"),query))
                .addHeader("Authorization", "Basic "+Authorization)
                .addHeader("Content-Type", "text/plain")
                .build();

        return request;
    }

    private void concurrentRead(int[] client) throws InterruptedException {
        Random ran = new Random();
        for(int i = 0; i < client.length; i++) {
            CountDownLatch CDL = new CountDownLatch(client[i]);
            AtomicLong AL = new AtomicLong(0);
            for(int k = 0; k < client[i]; k++) {
                new Thread(() ->{
                    long st = System.nanoTime();
                    Response response = TDEngine_Common.send(getRequest(ran.nextInt(bound), ran.nextInt(bound)));
                    assert response != null;
                    response.close();
                    AL.addAndGet((System.nanoTime() - st) / constant.Ns2Ms);
                    CDL.countDown();
                }).start();
            }

            CDL.await();
            System.out.printf("[TSBench-D] >>>>> [TDEngine_ConcurrentRead] ThreadNum:%d, average response time : %d ms>>>>> \n", client[i], AL.longValue() / (long)client[i]);
            IOUtils.WriteData(constant.GetWriteFilePath("TDEngine", "concurrentRead"),
                    new String[]{String.valueOf(client[i]), String.valueOf(AL.longValue() / (long)client[i])});
        }
    }

    public static void main(String[] args) throws IOException, InterruptedException {
        TDEngine_ConcurrentRead db = new TDEngine_ConcurrentRead(constant.IP, constant.TDEngine_port, constant.dbName, constant.Concurrent_tbName, constant.userName, constant.passWord, constant.bound);
        db.concurrentRead(constant.ConcurrentReadThreadNum);
    }
}
